import java.util.concurrent.Flow;
import java.util.concurrent.Flow.Subscription;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo2 {
    public static void main(String[] args) throws InterruptedException {
        //1.定义发布者，发布的数据类型是Integer
        //直接使用jdk自带的SubmissionPublisher
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        //2.定义处理器，对数据进行过滤，并转换为String类型
        MyProcessor processor = new MyProcessor();

        //3.发布者和处理器建立订阅关系
        publisher.subscribe(processor);

        //4.定义最终订阅者
        Subscriber<String> subscriber = new Subscriber<String>() {

            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                //保存订阅关系,需要用它来给发布者响应
                this.subscription = subscription;

                //请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                //接收到一个数据，处理
                System.out.println("接收到的数据:"+item);

                //处理完调用request再请求一个数据
                this.subscription.request(1);

                //
            }

            @Override
            public void onError(Throwable throwable) {

                //出现异常
                throwable.printStackTrace();
                //告诉发布者，后面不接收数据
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {

                //全部数据处理完了(发布者关闭了)

            }
        };

        //5.处理器 和 最终订阅者 建立订阅关系
        processor.subscribe(subscriber);

        //6.生产数据,并发布
        publisher.submit(-111);
        publisher.submit(111);
        publisher.submit(22222);

        //7.结束后 关闭发布者
        //正式环境 应该放在finally 或者使用 try-resouce确保关闭
        publisher.close();

        //主线程延迟停止，否则数据没有消费就退出
        Thread.currentThread().join(1000);

    }
}
